package tasks;

import java.io.FileReader;
import java.io.IOException;
import java.io.StreamTokenizer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import system.MapperResult;
import system.ReducerResult;

import api.TaskBase;

/**
 * MapReduce implementation of Inverted Index task
 * 
 * @author Mayuri Karra, Bing Wei
 *
 */
public class InvertedIndexTask extends TaskBase{

	private static final long serialVersionUID = 1L;

	/**
	 * Constructor 
	 * @param inputFile name of input file
	 * @param outputFile name of output file
	 */
	public InvertedIndexTask(String inputFile, String outputFile) {
		super(inputFile, outputFile);
	}

	/**
	 * Map method for inverted index. It takes a <key, value> pair
	 * and returns a list of MapperResult pairs
	 * @param key that is to mapped
	 * @param value 
	 * @return a list of MapperResults which are essentially <key,value> pairs 
	 */

	@Override
	public List<MapperResult> map(Object key, Object value) {
		// TODO Auto-generated method stub
		StreamTokenizer st = null;
	    List<MapperResult> keyValueList = new ArrayList<MapperResult>();
	    Integer docId = (Integer)value;

		try {
			st =  new StreamTokenizer(new FileReader((String)key));
			//System.out.println("reading: " + key);
	      Map<String, Integer> wordCount = new HashMap <String, Integer>();
	
	        while (st.nextToken() != StreamTokenizer.TT_EOF){
	        	if(st.ttype == StreamTokenizer.TT_WORD){
	        		String word = st.sval;
	        		if (wordCount.containsKey(word)){
	         			wordCount.put(word,wordCount.get(word) + 1);
	         		}else{
	         			wordCount.put(word, 1);
	         		}
	        	}
         		
        	}
	        
	        for(Map.Entry<String, Integer> entry: wordCount.entrySet()){
	        	DocAndCount docCount = new DocAndCount(docId, entry.getValue());
	        	keyValueList.add(new MapperResult(entry.getKey(), docCount.toString()));
	        }
        }
	    catch (IOException ex){
	      ex.printStackTrace();
	    }
	    
		
		//System.out.println("Done reading : " + key);
      return keyValueList;
		
	}

	/**
	 * Method for the reduce operation of inverted index task. 
	 * It takes a list of <key, list<value>> pairs
	 * and returns ReducerResult
	 * 
	 * @param key
	 * @param values
	 * @return ReducerResult which has <key, value>
	 */
	@Override
	public  ReducerResult reduce(Object key, List<Object> values) {
		List<DocAndCount> rsts = null;
		try{
			rsts = new ArrayList<DocAndCount>();
			for(Object obj: values){
				String[] splits = ((String)obj).split(":");
				DocAndCount dc = new DocAndCount(Integer.valueOf(splits[0]), Integer.valueOf(splits[1]));
				rsts.add(dc);
			}
			Collections.sort(rsts, new DocAndCountComparator());
		}catch(Exception e){
			;
		}
		return new ReducerResult(key, rsts);
		
	}
	
	/**
	 * Helper class used in Map/Reduce phase of inverted index
	 * 
	 * @author Mayuri Karra, Bing Wei
	 *
	 */
	class DocAndCount{
		int docId;
		int wordCount;
		
		public DocAndCount(int docId, int wordCount) {
			super();
			this.docId = docId;
			this.wordCount = wordCount;
		}
		
		public int getDocId() {
			return docId;
		}

		public int getWordCount() {
			return wordCount;
		}

		public String toString(){
			return docId + ":" + wordCount;
		}
	}
	/**
	 * Comparator implementation that enables sorting of DocAndCount class
	 * @author Mayuri Karra, Bing Wei
	 *
	 */
	class DocAndCountComparator implements Comparator<DocAndCount>{
		public int compare(DocAndCount t1, DocAndCount t2){
			if(t1.getWordCount()< t2.getWordCount())
				return 1;
			else if(t1.getWordCount()> t2.getWordCount())
				return -1;
			return 0;
		}
	}
}
